package mysrpringboot.springboot.rocketmq;


import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;

@Component
public class Myconsumer {
    public List consumer() throws MQClientException{
        List list = new ArrayList();
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_grp_04_01");
        consumer.setNamesrvAddr("192.168.159.162:9876");
        consumer.setConsumeMessageBatchMaxSize(1);
        consumer.setConsumeThreadMin(1);
        consumer.setConsumeThreadMax(1);

        consumer.subscribe("tp_Topic_001", "*");
        // 并发消费
        // consumer.setMessageListener(new MessageListenerConcurrently() {
        // @Override
        // public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context)
        // {
        // return null;
        // }
        // });
        // 顺序消费
        consumer.setMessageListener(
                new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(
                    List<MessageExt> msgs, ConsumeOrderlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.println(msg.getMsgId() + "\t" + msg.getQueueId() + "\t" + new String(msg.getBody()));
                    list.add(new String(msg.getBody()));
                }
                return null;
            }
        });

        consumer.start();
        return list;
    }

}
